package cn.doitedu.udfs;

import cn.doitedu.pojo.*;
import cn.doitedu.service.MatchService;
import cn.doitedu.utils.StateDescriptorUtils;
import com.starrocks.shade.com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.MeterView;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.kie.api.io.ResourceType;
import org.kie.api.runtime.KieSession;
import org.kie.internal.utils.KieHelper;

import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;

/**
 * MatchFunctionV4 是调用MatchServer，MatchFunctionV4相当于Controller
 *
 * Flink的状态分为两种：
 *  KeyedState：
 *      ValueState、MapState、ListState
 *  OperatorState（BroadCast State属于OperatorState）
 *      ListState
 *
 * 以后程序出现问题，希望从广播状态中恢复数据（RulesBean）RuleBean里面有json字符串（规则参数）、DRL字符串（运算逻辑）
 * 程序出现异常，可以恢复保存在广播状态中的json字符串（规则参数）、DRL字符串（运算逻辑），然后得到RulesCondition和KieSession
 */
public class MatchFunctionV5 extends KeyedBroadcastProcessFunction<String, LogBean, RulesBean, MatchResult> implements CheckpointedFunction {


    private RuleConditionV2 ruleCondition;

    private MatchService matchService;

    //程序出现异常，根据重启策略重启，那么rulesMap中的规则参数和规则逻辑都没有了
    private LinkedHashMap<Long, Tuple2<RuleConditionV2, KieSession>> rulesMap;

    //保存是对注册定时器是对应的触发时间和行为参数（未完成）
    private transient MapState<Long, TimerCondition> timerConditionMapState;

    //保存用户最近两个小时的行为
    private transient ListState<EventBean> listState;

    //缓存数据（为了减少对Doris的查询）
    private transient MapState<String, String> cacheState;

    private Counter allCounter;
    private Counter hitCounter;
    private MeterView meterView;

    //会在open方法执行之前调用一次
    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        rulesMap = new LinkedHashMap<>();
        //获取OperatorState（广播状态）
        BroadcastState<Long, RulesBean> broadcastState = context.getOperatorStateStore().getBroadcastState(StateDescriptorUtils.rulesStateDescriptor);
        for (Map.Entry<Long, RulesBean> entry : broadcastState.entries()) {
            Long id = entry.getKey();
            RulesBean rulesBean = entry.getValue();
            String conditionJson = rulesBean.getRule_condition_json();
            //将json字符串转成对象，为了提高效率
            RuleConditionV2 ruleCondition = JSON.parseObject(conditionJson, RuleConditionV2.class);
            //根据DRL文件构建KieSession
            String controllerDrl = rulesBean.getRule_controller_drl();
            //创建KieSession的工具类
            KieHelper kieHelper = new KieHelper();
            //添加指定规则的内容
            kieHelper.addContent(controllerDrl, ResourceType.DRL);
            //创建规则引擎会话(要不规则代码片段动态加载，动态编译)，KieSession创建比较耗时，一个规则最好使用一个KieSession
            KieSession kieSession = kieHelper.build().newKieSession();
            rulesMap.put(id, Tuple2.of(ruleCondition, kieSession));
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //在TaskManager初始化subtask时，获取全局参数
        ParameterTool parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();

        timerConditionMapState = getRuntimeContext().getMapState(StateDescriptorUtils.timerConditionStateDescriptor);
        //设置近期行为状态的TTL
        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(2)).build();
        ListStateDescriptor<EventBean> eventBeanStateDescriptor = StateDescriptorUtils.eventBeanStateDescriptor;
        eventBeanStateDescriptor.enableTimeToLive(ttlConfig);

        listState = getRuntimeContext().getListState(eventBeanStateDescriptor);

        StateTtlConfig ttlConfig2 = StateTtlConfig.newBuilder(Time.hours(6)).build();
        MapStateDescriptor<String, String> cacheStateDescriptor = StateDescriptorUtils.cacheStateDescriptor;
        cacheStateDescriptor.enableTimeToLive(ttlConfig2);
        cacheState = getRuntimeContext().getMapState(cacheStateDescriptor);

        //初始化MatchServer
        matchService = new MatchService();
        matchService.init(parameterTool, listState, cacheState);

        //统计条数
        MetricGroup metricGroup1 = getRuntimeContext().getMetricGroup().addGroup("g1");
        allCounter = metricGroup1.counter("all");
        hitCounter = metricGroup1.counter("hit");


        //统计速率的
        MetricGroup metricGroup2 = getRuntimeContext().getMetricGroup().addGroup("g1");

        meterView = metricGroup2.meter("hit-rate", new MeterView(2));
    }


    @Override
    public void processElement(LogBean bean, ReadOnlyContext ctx, Collector<MatchResult> out) throws Exception {
        allCounter.inc();
        //将用户的行为数据保存到状态中（近期数据）
        listState.add(bean.toEventBean());

        for (Map.Entry<Long, Tuple2<RuleConditionV2, KieSession>> entry : rulesMap.entrySet()) {
            Long id = entry.getKey();
            Tuple2<RuleConditionV2, KieSession> tp = entry.getValue();
            RuleConditionV2 ruleCondition = tp.f0;
            FactBean factBean = new FactBean(bean, ruleCondition, matchService, false);
            KieSession kieSession = tp.f1;
            kieSession.insert(factBean);
            kieSession.fireAllRules();
            if (factBean.isMatch()) {

                //不带定时任务的规则
                if (!ruleCondition.isHasTimer()) {
                    //返回结果
                    out.collect(new MatchResult(bean.getDeviceId(), bean.getTimeStamp(), id.toString(), System.currentTimeMillis()));
                    hitCounter.inc();
                    meterView.markEvent();
                } else {
                    //带定时任务的规则
                    //注册定时器，在指定的时间后，用定时器检测是否未完成某一件事情
                    TimerCondition timerCondition = ruleCondition.getTimerCondition();
                    //注册定时器
                    Long lateTime = timerCondition.getLateTime();
                    //使用EventTime
                    //long currentTime = System.currentTimeMillis();
                    long triggerTime = bean.getTimeStamp() + lateTime;
                    ctx.timerService().registerProcessingTimeTimer(triggerTime);

                    //注册的的定时器对应的行为参数（未完成的）
                    timerConditionMapState.put(triggerTime, timerCondition);
                }
            }
        }

    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<MatchResult> out) throws Exception {

        //检测一定时间之后，是否未完成某个行为或行为序列
        //获取事先注册定时器对应的行为参数呢？
        TimerCondition timerCondition = timerConditionMapState.get(timestamp);
        if (timerCondition != null) {
            //查询是否为完成某个行为
            LogBean logBean = new LogBean();
            String deviceId = ctx.getCurrentKey();
            logBean.setDeviceId(deviceId);
            boolean isDone = matchService.isMatchEventCount(logBean, Collections.singletonList(timerCondition.getCombineCondition()));
            if(!isDone) {
               //发送消息到消息中间件
                out.collect(new MatchResult(deviceId, timestamp - timerCondition.getLateTime() ,timerCondition.getId() + "",  timestamp));
            }
        }

    }

    //job出现异常重启后，FlinkCDC会重新读取MySQL中的规则数据，还是接着以前的偏移量继续读？
    //开启Checkpoint后，FlinkCDC读取MySQL的数据，会将偏移量保存到状态中
    @Override
    public void processBroadcastElement(RulesBean rulesBean, Context ctx, Collector<MatchResult> out) throws Exception {
        //将动态的规则保存到广播状态中
        BroadcastState<Long, RulesBean> broadcastState = ctx.getBroadcastState(StateDescriptorUtils.rulesStateDescriptor);
        String type = rulesBean.getType();
        Long id = rulesBean.getId();
        if (!"d".equals(type)) {
            //RulesBean保存到了广播状态中了
            broadcastState.put(id, rulesBean);
            //KieSession不支持序列化，没法保存到StateBackend中，checkpoint是没法保存
            //所有KieSession没法保存到广播状态中
            //RuleConditionV2（规则参数），KieSession（运算逻辑）
            String conditionJson = rulesBean.getRule_condition_json();
            //将json字符串转成对象，为了提高效率
            RuleConditionV2 ruleCondition = JSON.parseObject(conditionJson, RuleConditionV2.class);
            //根据DRL文件构建KieSession
            String controllerDrl = rulesBean.getRule_controller_drl();
            //创建KieSession的工具类
            KieHelper kieHelper = new KieHelper();
            //添加指定规则的内容
            kieHelper.addContent(controllerDrl, ResourceType.DRL);
            //创建规则引擎会话(要不规则代码片段动态加载，动态编译)，KieSession创建比较耗时，一个规则最好使用一个KieSession
            KieSession kieSession = kieHelper.build().newKieSession();

            //将规则参数，和运算逻辑，保存到map结合中
            rulesMap.put(id, Tuple2.of(ruleCondition, kieSession));
        } else {
            broadcastState.remove(id);
            rulesMap.remove(id);
        }

    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {}

    @Override
    public void close() throws Exception {
        //释放资源
        matchService.close();
    }
}
